import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;

class TopUser extends Configured implements Tool 
{	
	public static class Map extends
			Mapper<LongWritable, Text, LongWritable, LongWritable> {

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			Long objID = Long.parseLong(tokenizer.nextToken());
			Long nTrust = Long.parseLong(tokenizer.nextToken());
			context.write(new LongWritable(nTrust), new LongWritable(objID));
		}
	}

	public static class Reduce extends
			Reducer<LongWritable, LongWritable, LongWritable, LongWritable> {
		public void reduce(LongWritable key, Iterable<LongWritable> values,
				Context context) throws IOException, InterruptedException {
				for (LongWritable val : values)
					context.write(key, new LongWritable(val.get()));
		}
	}

	public int run(String[] args) throws Exception {
		/* 
		 * Calculate ratings per article
		 */
		Job job = new Job(getConf());
		job.setJarByClass(TopUser.class);
		job.setJobName("TopUser");

		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(LongWritable.class);

		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		boolean success = job.waitForCompletion(true);
		return success ? 0 : 1;
	}
 
	 public static void callMR(String[] args) throws Exception {
		int ret = ToolRunner.run(new TopUser(), args);
		System.exit(ret);
	}
}

class TopUserArray {
	public static ArrayList<Long> topUser = new ArrayList<Long> (100);
	
		public static void readTopUser(String file) throws IOException {
			Scanner scan = new Scanner(new File(file));
			while (scan.hasNextLine()) {
				String line = scan.nextLine();
				StringTokenizer token = new StringTokenizer(line);
				token.nextToken();
				topUser.add(Long.parseLong(token.nextToken()));
			}
		}
}

class TopRatedArticleByTopUser extends Configured implements Tool  {
	
	private static Long member = 0L;
	
	public TopRatedArticleByTopUser(){}
	
	public TopRatedArticleByTopUser(Long memberId) {
		member = memberId;
	}
	
	public static class Map extends
	Mapper<LongWritable, Text, LongWritable, LongWritable> {
		public void map(LongWritable key, Text value, Context context)
		throws IOException, InterruptedException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			Long objID = Long.parseLong(tokenizer.nextToken());
			Long memberId = Long.parseLong(tokenizer.nextToken());
			Long rating = Long.parseLong(tokenizer.nextToken());
			tokenizer.nextToken();
			Text date =new Text(tokenizer.nextToken());
			
			if(rating == 5 && member == memberId) {
				context.write(new LongWritable(objID), new LongWritable(1));
			}
		}	
	}

	public static class Reduce extends
		Reducer<LongWritable, LongWritable, LongWritable, LongWritable> {
		public void reduce(LongWritable key, Iterable<LongWritable> values,
		Context context) throws IOException, InterruptedException {
			Long sum = 0L;
			for (LongWritable val : values) {
				sum += val.get();
			}
			context.write(key, new LongWritable(sum));
		}
	}

	public int run(String[] args) throws Exception {
	/* 
	 * Calculate ratings per article
	 */
		Job job = new Job(getConf());
		job.setJarByClass(TopRatedArticleByTopUser.class);
		job.setJobName("TopRatedArticleByTopUser");

		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(LongWritable.class);

		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		boolean success = job.waitForCompletion(true);
		return success ? 0 : 1;
	}

	public void callMR(String file, Long memberId) throws Exception {
		String []files = new String[2];
		files[0] = file;
		files[1] = memberId.toString();
		int ret = ToolRunner.run(new TopRatedArticleByTopUser(memberId), files);
		System.exit(ret);
	}
}

public class Influence {
	public static void main(String[] args) throws Exception {
		//TopUser.callMR(args);
		TopUserArray.readTopUser("TopUser.txt/TopUser.txt");
		for (int i = 0; i < 100; i++) {
			TopRatedArticleByTopUser tA = new TopRatedArticleByTopUser();
			tA.callMR("rating.txt", TopUserArray.topUser.get(i));
		}	
	}
}